package com.sarnath.sardoop.modules.hbase.service;

import javax.annotation.Resource;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Service;

@SuppressWarnings("deprecation")
@Service("hbaseDdl")
@Lazy(false)
public class HbaseDdlService {
	@SuppressWarnings("unused")
	private Logger logger=LoggerFactory.getLogger(getClass());
	@Resource
	private HbaseTemplate htemplate;
	
	public void createTable(String tableName,String... familys){
		Configuration conf=htemplate.getConfiguration();
		try(Connection conn=ConnectionFactory.createConnection(conf)){
			HBaseAdmin admin = (HBaseAdmin) conn.getAdmin(); 
			if(admin.tableExists(tableName)){
	        	throw new RuntimeException("表已经存在！");
	        }else{
	        	HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
	        	for(String f:familys){
	        		HColumnDescriptor coldef = new HColumnDescriptor(f);
		        	coldef.setCompressionType(Algorithm.GZ);//列族压缩格式
		        	coldef.setDataBlockEncoding(DataBlockEncoding.PREFIX_TREE);
		            desc.addFamily(coldef);//添加列族	
	        	}
	            admin.createTable(desc);
	        }
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	public void dropTable(String tableName){
		Configuration conf=htemplate.getConfiguration();
		try(Connection conn=ConnectionFactory.createConnection(conf)){
			HBaseAdmin admin = (HBaseAdmin) conn.getAdmin(); 
			if(!admin.tableExists(tableName)) throw new RuntimeException("表不存在！");
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	public void truncate(String tableName){
		Configuration conf=htemplate.getConfiguration();
		try(Connection conn=ConnectionFactory.createConnection(conf)){
			HBaseAdmin admin = (HBaseAdmin) conn.getAdmin(); 
			if(!admin.tableExists(tableName)) throw new RuntimeException("表不存在！");
			admin.disableTable(tableName);
			admin.truncateTable(TableName.valueOf(tableName), true);
			//admin.enableTable(tableName);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
	
	public void insert(final String tableName,final String family,final String[] rowkeys,final String[] fields,final String[] values){
		htemplate.execute(tableName,  new TableCallback<Object>() {
			@Override
			public Object doInTable(HTableInterface table) throws Throwable {
				for(int i=0;i<rowkeys.length;i++){
					  Put p = new Put(Bytes.toBytes(rowkeys[i]));
					    p.add(Bytes.toBytes(family), Bytes.toBytes(fields[i]), Bytes.toBytes(values[i]));
					    table.put(p);
				  }
			    return null;
			}
			});
	}
	
	public void delete(String tableName,String familyName,String fieldName,String... rowkeys){
		if(fieldName==null){
			for(String rowkey:rowkeys) htemplate.delete(tableName, rowkey, familyName);
		}else{
			for(String rowkey:rowkeys) htemplate.delete(tableName, rowkey, familyName, fieldName);
		}
	}
}
